package com.example.shop.merchant.websocket;

import com.example.shop.common.config.InformWebSocketConfigurator;
import com.example.shop.common.enums.SessionAttributeKey;
import com.example.shop.common.enums.WebSocketUserPropertiesKey;
import com.example.shop.pojo.Merchant;
import com.example.shop.rabbitmq.component.MerchantMessageComponent;

import com.example.shop.rabbitmq.dto.ChatMessage;
import com.example.shop.redis.enums.RedisKey;
import com.example.shop.redis.service.LettuceClient;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.ConsumerShutdownSignalCallback;
import com.rabbitmq.client.DeliverCallback;
import org.elasticsearch.common.collect.CopyOnWriteHashMap;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.session.data.redis.RedisIndexedSessionRepository;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

@Component
@ServerEndpoint(value = "/websocket/chatMessage/merchant/{key}",configurator = InformWebSocketConfigurator.class)
public class MessageWebSocketServer {

    /**
     * 当前建立websocket连接的商家人数
     */
    private static int onlineCount = 0;
    /**
     * 所有建立websocket连接的会话map
     */
    private static CopyOnWriteHashMap<String, MessageWebSocketServer> webSocketMap = new CopyOnWriteHashMap<>();
    /**
     * 建立websocket的session会话
     */
    private Session session;
    private static ObjectMapper objectMapper;
    private static LettuceClient redisClient;
    private static RedisIndexedSessionRepository sessionRepository;
    private static RabbitTemplate rabbitTemplate;
    /**
     * 队列名
     */
    private String queueName;
    /**
     * 消费者标签
     */
    private String consumerTag;

    @Autowired
    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        if(MessageWebSocketServer.rabbitTemplate == null){
            MessageWebSocketServer.rabbitTemplate = rabbitTemplate;
        }
    }

    @Autowired
    public void setRedisClient(LettuceClient redisClient) {
        if(MessageWebSocketServer.redisClient == null){
            MessageWebSocketServer.redisClient = redisClient;
        }
    }

    @Autowired
    public void setObjectMapper(@Qualifier("commonObjectMapper") ObjectMapper objectMapper) {
        if(MessageWebSocketServer.objectMapper == null){
            MessageWebSocketServer.objectMapper = objectMapper;
        }
    }

    @Autowired
    public void setSessionRepository(RedisIndexedSessionRepository sessionRepository) {
        if(MessageWebSocketServer.sessionRepository == null){
            MessageWebSocketServer.sessionRepository = sessionRepository;
        }
    }

    /**
     * 根据websocket的session会话id来获取该MessageWebSocketServer
     * @param sessionId websocket的session会话id
     * @return 返回MessageWebSocketServer对象
     */
    public static MessageWebSocketServer getWebSocketServer(String sessionId){
        return webSocketMap.get(sessionId);
    }

    @OnOpen
    public void onOpen(Session session,@PathParam("key") String key) throws IOException {
        //当前建立websocket连接的商家人数+1
        incrOnlineCount();

        Map<String, Object> userProperties = session.getUserProperties();
        //获取websocket的ip
        String connIp = (String) userProperties.get(WebSocketUserPropertiesKey.CLIENT_IP.getKey());
        //获取请求令牌时关联的HttpSession的id
        String reqTokenHttpSessionId = (String) redisClient.get(RedisKey.MERCHANT_WEBSOCKET_TOKEN_REQ_SESSION.getKey() + key);
        //如果从redis中获取不到websocket令牌，说明没有先去请求获取令牌
        if(reqTokenHttpSessionId == null){
            session.close(new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT,"不存在该令牌，请先去请求websocket令牌"));
            return;
        }
        //获取请求令牌时关联的HttpSession
        org.springframework.session.Session reqTokenHttpSession =  sessionRepository.findById(reqTokenHttpSessionId);
        //判断请求令牌时的ip是否与当前要建立websocket的ip一致
        String reqIp = reqTokenHttpSession.getAttribute(SessionAttributeKey.CLIENT_IP.getKey());
        if(!connIp.equals(reqIp)){
            session.close(new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT,"请求websocket令牌与建立连接不是同一端"));
            return;
        }
        //符合建立连接，删除websocket令牌
        redisClient.delKey(RedisKey.MERCHANT_WEBSOCKET_TOKEN_REQ_SESSION.getKey() + key);

        //设置二进制消息最大1M
        session.setMaxBinaryMessageBufferSize(8*1024*1024);
        //设置此会话的空闲超时时间。毫秒为单位。零或负值表示无限超时。
        session.setMaxIdleTimeout(1000*60*3);
        //设置文本消息，最大8*1024个字符
        session.setMaxTextMessageBufferSize(8*1024);
        this.session = session;
        webSocketMap = webSocketMap.copyAndPut(session.getId(),this);

        //获取当前session中登录的用户
        Merchant merchant = reqTokenHttpSession.getAttribute(SessionAttributeKey.NOW_LOGIN_MERCHANT.getKey());
        //商家的消息队列名
        queueName = MerchantMessageComponent.QUEUE_PRE + merchant.getMerchantId();
        //将该商家中消息队列的所有消息都发送给前端
        //如果该队列已经有消费者了，则说明该队列在其他地方已经连接上过了
        String consumerTag = rabbitTemplate.execute(channel -> {

            long l = channel.consumerCount(queueName);
            if (l != 0) {
                return null;
            }
            DeliverCallback deliverCallback = (cTag, publishMessage) -> {
                //获取消息内容
                String json = new String(publishMessage.getBody(), StandardCharsets.UTF_8);
                //将该消息内容转换为ChatMessage对象
                ChatMessage chatMessage = objectMapper.readValue(json, ChatMessage.class);
                try {
                    //发送消息给前端,如果当发送消息的过程中，前端突然中断，则可能会报错，因为异步close(),session变为null,但这边正在发消息,
                    sendMessage(chatMessage);
                    channel.basicAck(publishMessage.getEnvelope().getDeliveryTag(),false);
                } catch (Exception e) {
                    e.printStackTrace();
                    channel.basicReject(publishMessage.getEnvelope().getDeliveryTag(),true);
                }
            };
            //消费者取消回调该接口
            CancelCallback cancelCallback = cTag -> onClose();
            //消费者关闭回调该接口
            ConsumerShutdownSignalCallback consumerShutdownSignalCallback = (cTag, sig) -> onClose();
            //流量控制
            channel.basicQos(10, false);
            return channel.basicConsume(queueName, false, deliverCallback, cancelCallback, consumerShutdownSignalCallback);
        });
        if(consumerTag == null){
            session.close(new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT,"该用户已经在其他地方进行websocket连接"));
            return;
        }
        //设置当前的消费者
        this.consumerTag = consumerTag;
    }

    /**
     * 接收客户端发送过来的消息
     * @param message json消息
     * @param session tcp连接会话
     */
    @OnMessage
    public void acceptMessage(String message,Session session){
        System.out.println("前端会话sessionId:"+session.getId()+" 消息："+message);
    }

    /**
     * 此时已经关闭session，即session为null
     */
    @OnClose
    public void onClose(){
        //立即取消该消费者消费
        consumerTag = rabbitTemplate.execute(channel -> {
            if(consumerTag!=null){
                channel.basicCancel(consumerTag);
            }
            return null;
        });
        Set<Map.Entry<String, MessageWebSocketServer>> entrySet = webSocketMap.entrySet();
        Iterator<Map.Entry<String, MessageWebSocketServer>> iterator = entrySet.iterator();
        while(iterator.hasNext()){
            Map.Entry<String, MessageWebSocketServer> next = iterator.next();
            if(next.getValue().equals(this)){
                webSocketMap = webSocketMap.copyAndRemove(next.getKey());
                break;
            }
        }
        decrOnlineCount();
    }

    @OnError
    public void onError(Session session,Throwable error){
        System.out.println("出错："+session.getId());
        rabbitTemplate.execute(channel -> {
            if(consumerTag!=null){
                channel.basicCancel(consumerTag);
            }
            return null;
        });
        error.printStackTrace();
    }


    public void sendMessage(Object obj) throws EncodeException, IOException {
        RemoteEndpoint.Basic basicRemote = session.getBasicRemote();
        basicRemote.sendText(objectMapper.writeValueAsString(obj));
    }



    synchronized public static  void incrOnlineCount(){
        onlineCount++;
    }

    synchronized public static  void decrOnlineCount(){
        onlineCount--;
    }
}
